package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.common.utils.HuoBiGZipUtils;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * biki websocket api
 * https://github.com/code-biki/open-api#websocket-api
 * @since  2020/3/31 15:40
 * @author luowenliang
 */
@Slf4j
@Component
public final class BiKiApplication extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

	private static final String WS_URL = "wss://ws.biki.com/kline-api/ws";

	private static final String DEPTH_CHANNEL = "market_%s_depth_step0";

	/**
	 * 执行心跳的间隔时间和在此基础商上的超时时间将由实现的子类所设定(或者以配置的方式来是定)
	 * 假如：此时间有父类适配器所设定那么所有共同子类将继承，如需改变其中一个子类的时间，将会造成
	 * 所有子类间隔时间更改
	 */
	private long period = 3000;


	@Autowired
	private SymbolService symbolService;

	@Override
	public void afterPropertiesSet() throws Exception {
		super.afterPropertiesSet(new BiKiWebSocketClient(new URI(WS_URL), new Draft_6455(), null, 5000));
		scheduled.scheduleWithFixedDelay(() -> heartBeat(), period, period, TimeUnit.MILLISECONDS);
	}

	@Override
	public void destroy() throws Exception {
		this.client.close();
		scheduled.shutdown();
	}


	/**
	 * {
	 *     "event": "sub",
	 *     "params": {
	 *         "channel": "market_$base$quote_depth_step[0-2]",
	 *         "cb_id": "自定义",
	 *         "asks": 150,
	 *         "bids": 150
	 *     }
	 * }
	 * $base$quote表示btcusdt等币对,step[0-2]表示深度有3个维度，0、1、2
	 * channel 示例 "channel": "market_btcusdt_depth_step0"
	 * @param symbol
	 */
	@Override
	public void subscribe(Symbol symbol) {
		//订阅事件
		JSONObject object = new JSONObject();
		JSONObject params = new JSONObject();
		object.put("event", "sub");

		params.put("channel", String.format(DEPTH_CHANNEL, BourseEnum.HUOBI.getSymbolFormat().symbolFormat(symbol)));
		object.put("cb_id", sequence.nextId());
		object.put("asks", 150);
		object.put("bids", 150);

		object.put("params", params);
		this.client.send(object.toJSONString());
	}

	/*public void subscribe(String symbol) {
		//订阅事件
		JSONObject object = new JSONObject();
		JSONObject params = new JSONObject();
		object.put("event", "sub");

		params.put("channel", String.format(DEPTH_CHANNEL, symbol));
		object.put("cb_id", sequence.nextId());
		object.put("asks", 150);
		object.put("bids", 150);

		object.put("params", params);
		this.client.send(object.toJSONString());
	}*/

	@Override
	public void unSubscribe(Symbol symbol) {
		cleanSymbolPrice(symbol.getName());
	}

	@Override
	public void subscribeList(Set<Symbol> subSet) {
		for (Symbol symbol: subSet) {
			subscribe(symbol);
		}
	}

	@Override
	public BoursePrice getSymbolPrice (String symbol) {
		return symbol_map.get(symbol);
	}

	@Override
	public void cleanSymbolPrice(String symbol) {
		symbol_map.remove(symbol);
	}

	private final class BiKiWebSocketClient extends WebSocketClient {

		public BiKiWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders,
				int connectTimeout) {
			super(serverUri, protocolDraft, httpHeaders, connectTimeout);
		}

		@Override
		public void onOpen(ServerHandshake handshakedata) {
			log.info("BiKi WebSocket接口: {} 链接成功.", this.getURI());
			Set<Symbol> subSet = symbolService.listByBourse(BourseEnum.BIKI.getValue());
			if (subSet != null && !subSet.isEmpty()) {
				for (Symbol symbol : subSet) {
					subscribe(symbol);
				}
			}
		}

		@Override
		public void onMessage(ByteBuffer bytes) {
			byte[] array = bytes.array();
			array = HuoBiGZipUtils.decompress(array);
			String str = new String(array, StandardCharsets.UTF_8);
			// log.info("收到BiKi websocket消息： {}", str);
			if(str.contains(PING)) {
				this.send(str.replace(PING, PONG));
			}else if(str.contains("tick")){
				JSONObject object = JSON.parseObject(str);
				JSONObject data = object.getJSONObject("tick");
				String channel = object.getString("channel");
				String symbol = channel.split("_")[1];
				if ( data != null) {
					JSONArray asks = data.getJSONArray("asks").getJSONArray(0);
					JSONArray buys = data.getJSONArray("buys").getJSONArray(0);
//					log.info("BiKi {}, 买一价：{}, 卖一价：{}", symbol, buys.getString(0), asks.getString(0));
					BoursePrice price = BoursePrice.builder().buyPrice(buys.getBigDecimal(0)).sellPrice(asks.getBigDecimal(0)).data(data.toJSONString()).updateTime(new Date()).build();
					// log.info("BiKi获取到【***{}***】的实时价格为：{}", symbol, price.toString());
					symbol_map.put(symbol, price);
				}
			}else {
				log.error("BiKi WebSocket数据未做处理");
			}
		}

		@Override
		public void onMessage(String message) {
			log.info("接收到BiKi的信息：{}",message);
		}

		@Override
		public void onClose(int code, String reason, boolean remote) {
			log.info("BiKi Websocket接口已关闭, 原因: {}, 状态: {}", reason, code);
		}

		@Override
		public void onError(Exception e) {
			log.error("BiKi WebSocket接口，发生异常", e);
		}
	}

}
